package cn.doitedu.demo.join;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Lookup关联 {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 创建连接器表，映射kafka中的ss-1 : 行为日志
        tenv.executeSql("CREATE TABLE user_events          (\n" +
                "  uid int                                \n" +
                "  ,event_id string                           \n" +
                "  ,action_time bigint                        \n" +
                "  ,properties map<string,string>             \n" +
                "  ,rt as to_timestamp_ltz(action_time,3)     \n" +
                "  ,watermark for rt as rt                    \n" +
                "  ,pt as proctime()                          \n" +
                ") WITH (                                    \n" +
                "  'connector' = 'kafka',                    \n" +
                "  'topic' = 'a-1',                         \n" +
                "  'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "  'properties.group.id' = 'doit44_g2',      \n" +
                "  'scan.startup.mode' = 'latest-offset',    \n" +
                "  'value.format' = 'json',                  \n" +
                "  'value.fields-include' = 'EXCEPT_KEY'     \n" +
                ")");



        // 创建jdbc连接器表，映射mysql中的 user_info：用户注册信息表
        tenv.executeSql("CREATE TABLE user_info_mysql       ( \n" +
                "  uid int,                                     \n" +
                "  gender  string,                              \n" +
                "  age  int                                    \n" +
                //"  primary key(uid) not enforced                \n" +
                ") WITH (                                       \n" +
                "   'connector' = 'jdbc',                       \n" +
                "   'url' = 'jdbc:mysql://doitedu:3306/flinktest',\n" +
                "   'table-name' = 'user_info',                   \n" +
                "   'username' = 'root',                      \n" +
                "   'password' = 'root'                       \n" +
                ");");



        // 写一个lookup关联的sql
        tenv.executeSql(
                "create temporary view wide_table as " +  // 临时视图：就是封装了一段sql逻辑
                 "select      \n" +
                "    a.uid,   \n" +
                "    a.event_id,\n" +
                "    a.action_time,\n" +
                "    a.properties,\n" +
                "    a.rt,        \n" +
                "    b.gender,    \n" +
                "    b.age        \n" +
                "from user_events  a  \n" +
                "join user_info_mysql FOR SYSTEM_TIME AS OF a.pt as b  \n" +
                "on a.uid = b.uid").print();


        // 练习： 对关联好的结果，统计每5分钟、各性别、每个页面上的uv和pv
        /* *
         *
         * +----+-------------+--------------------------------+----------------------+--------------------------------+--------------------------------+-------------+
           | op |         uid |                       event_id |          action_time |                     properties |                         gender |         age |
           +----+-------------+--------------------------------+----------------------+--------------------------------+--------------------------------+-------------+
           | +I |           1 |                      page_load |        1704719574000 |               {ref=/x, url=/a} |                           male |          18 |
           | +I |           1 |                       add_cart |        1704719574000 |          {item_id=p01, url=/a} |                           male |          18 |
           | +I |           1 |                      page_load |        1704719574000 |               {ref=/a, url=/b} |                           male |          18 |
           | +I |           2 |                      page_load |        1704719574000 |               {ref=/x, url=/a} |                         female |          22 |
           | +I |           2 |                     video_play |        1704719574000 |         {video_id=v01, url=/a} |                         female |          22 |
           | +I |           2 |                      page_load |        1704719574000 |               {ref=/x, url=/b} |                         female |          22 |
           | +I |           1 |                      page_load |        1704719574000 |               {ref=/x, url=/c} |                           male |          18 |
           | +I |           1 |                     item_share |        1704719574000 |          {item_id=p03, url=/a} |                           male |          18 |
         *
         */
        tenv.executeSql(
                "select\n" +
                "    window_start,\n" +
                "    window_end,\n" +
                "    gender,\n" +
                "    properties['url'] as url,\n" +
                "    count(distinct uid) as uv,\n" +
                "    count(1) filter(where event_id ='page_load') as pv\n" +
                "from table(\n" +
                "tumble(table wide_table,descriptor(rt),interval '5' minute)\n" +
                ")\n" +
                "group by \n" +
                "    window_start,\n" +
                "\twindow_end,\n" +
                "\tgender,\n" +
                "\tproperties['url']").print();



    }

}
